import abc
import dataclasses
import math
import multiprocessing
import time
from typing import Any, Dict, List

import grpc
import pydash
from dacite import from_dict

from ys_python.common.base.config import get_config_value
from ys_python.common.base.logger import get_logger_with_cfg, LoggerCfg
from ys_python.common.base.single_ton import SingleTon
from ys_python.proto import MQ_pb2_grpc, MQ_pb2
from .ys_queue import IYSQueue, SimpleYSQueue, GRPCYSQueue


class IMQ(abc.ABC):
    """
    消息队列虚类
    """

    @abc.abstractmethod
    def subscribe(self, topics: str) -> IYSQueue:
        """订阅

        订阅一个话题

        Args:
            topics: 话题名 一次可以订阅多个话题名，用逗号分隔

        Returns:
            队列
        """
        pass

    @abc.abstractmethod
    def publish(self, topic: str, msg: Any) -> int:
        """
        发布消息到指定话题
        :param topic: 话题名 唯一
        :param msg: 消息, 任意类型数据与话题要求一致即可
        :return: -1 发送失败  0+ 消息被接收的数量
        """
        pass

    @abc.abstractmethod
    def unsubscribe(self, queue: IYSQueue) -> bool:
        """
        取消订阅
        :param queue: 已订阅的队列
        :return: True 取消成功 False 取消失败
        """
        pass


@SingleTon
class SimpleMQ(IMQ):
    """
    简单消息队列
    在不启动进程的情况下可以使用此类进行MQ 消息传递
    """

    __sub_dict: Dict[str, List[IYSQueue]] = {}  # 订阅管理列表
    __id = 0

    __pipe = multiprocessing.Pipe()

    def subscribe(self, topics: str) -> IYSQueue:

        if topics not in self.__sub_dict:
            self.__sub_dict[topics] = []

        _queue = SimpleYSQueue(self.__id)
        self.__id += 1
        self.__sub_dict[topics].append(_queue)
        return _queue

    def publish(self, topic: str, msg: Any) -> int:

        # 利用正则表达式看是否是订阅的话题
        _topics = self.__sub_dict.keys()
        _send_topics = []
        for _topic in _topics:
            if _topic.startswith(topic):
                _send_topics.append(_topic)

        # 对匹配的话题发送数据
        _total = 0
        for _topic in _send_topics:
            for _queue in self.__sub_dict[_topic]:
                if _queue.try_put(msg):
                    _total += 1

        return _total

    def unsubscribe(self, queue: IYSQueue) -> bool:
        _queue_list = self.__sub_dict[queue.topic]
        _queue_list.remove(queue)
        del queue
        return True


# region GRPC MQ 配置
_GRPC_MQ_CFG_NODE_NAME = "ys.grpc"

MQ_TOPIC_STATE = "json.state"  # 通用状态话题
MQ_TOPIC_HEART = "json.state"  # 通用状态话题
MQ_TOPIC_CFG = "json.cfg"  # 通用配置话题
MQ_TOPIC_OK = "json.ok"  # 通用返回话题
MQ_TOPIC_DATA = "json.data"  # 通用数据返回


@dataclasses.dataclass
class GRPCMQCfg:
    host: str = "127.0.0.1:10101"
    size: int = 1024 * 1024 * 3  # 3M分包
    logger: LoggerCfg = LoggerCfg()


@SingleTon
class GRPCMQ(IMQ):
    __id = 0
    __cfg: GRPCMQCfg

    def __init__(self):
        self.__init_cfg()
        self.__logger = get_logger_with_cfg(self.__cfg.logger, "mq")
        self.__logger.info("MQ配置：{}".format(self.__cfg))
        self.__channel = grpc.insecure_channel(self.__cfg.host)
        self.__stub = MQ_pb2_grpc.MQStub(self.__channel)

    def __init_cfg(self):
        dictionary = get_config_value("$.{}".format(_GRPC_MQ_CFG_NODE_NAME))
        if dictionary is None:
            self.__cfg = GRPCMQCfg()
        else:
            self.__cfg = from_dict(data_class=GRPCMQCfg, data=dictionary)

    def subscribe(self, topics: str) -> IYSQueue:
        self.__logger.info("订阅: " + topics)
        # 默认启动时 id 为 -1
        return GRPCYSQueue(-1, self.__stub, topics, self.__cfg.logger)

    def publish(self, topic: str, msg: bytes) -> int:
        """
        发送 MQ 消息
        Args:
            topic: 推送的话题
            msg: 消息信息

        Returns:

        """
        try:
            self.__logger.info("发布: " + topic)
            self.__logger.debug("消息:\n{}".format(str(msg)))
            # 消息分包 根据 size 大小对数据进行分包，包要拼接之后才能使用
            msg_len = len(msg)
            if msg_len > self.__cfg.size:
                packs = math.ceil(msg_len / self.__cfg.size)
                self.__logger.info("消息长度为: {}，分包数: {}".format(msg_len, packs))
                msgs = pydash.chunk(msg, self.__cfg.size)
                num = 0
                for i in range(len(msgs)):
                    pack_topic = topic + "[{}-{}]".format(i, packs)
                    self.__logger.info("正在发送：{}".format(pack_topic))
                    res = self.__stub.Pub(MQ_pb2.Message(topic=pack_topic, message=msgs[i]))
                    num += res.num
                    time.sleep(0.02)
                return num
            # 消息不大时直接发送
            else:
                res = self.__stub.Pub(MQ_pb2.Message(topic=topic, message=msg))
                return res.num
        except grpc.RpcError as e:
            self.__logger.warning("发布失败!\n", e.__str__())
            return 0

    def unsubscribe(self, queue: IYSQueue) -> bool:
        self.__stub.UnSub(MQ_pb2.UnSubRequest(id=queue.id))
        del queue
        return True
# endregion
